package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo17ForeachPartition {
  def main(args: Array[String]): Unit = {
    // 统计班级人数，将最终的结果写入MySQL

    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo17ForeachPartition")

    val sc: SparkContext = new SparkContext(conf)

    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")

    val clazzCntRDD: RDD[(String, Int)] = stuRDD.map(line => (line.split(",")(4), 1))
      .reduceByKey(_ + _)


    //    clazzCntRDD
    //      .foreach(kv => {
    //        /**
    //         * 同map算子类似，无法将连接建立在算子外部，否则会出现Task不能够被序列化的问题
    //         * 同样的放到算子内部则性能太低了，每一条数据会建立一次连接
    //         */
    //        // 建立JDBC连接
    //        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student?characterEncoding=utf8", "root", "123456")
    //        // 创建prepareStatement
    //        val pSt: PreparedStatement = conn.prepareStatement("insert into clazz_cnt (clazz,cnt) values (?,?)")
    //
    //
    //        val clazz: String = kv._1
    //        val cnt: Int = kv._2
    //
    //        // 设置参数
    //        pSt.setString(1, clazz)
    //        pSt.setInt(2, cnt)
    //
    //        // 执行插入语句
    //        pSt.execute()
    //
    //      })

    clazzCntRDD
      // 同mapPartitions类似 适用于将结果写到外部系统
      .foreachPartition(iter => {
        // 建立JDBC连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student?characterEncoding=utf8", "root", "123456")
        // 创建prepareStatement
        val pSt: PreparedStatement = conn.prepareStatement("insert into clazz_cnt (clazz,cnt) values (?,?)")

        var c: Int = 0
        val batchSize: Int = 1000
        // 处理每个分区内部的每一条数据
        iter
          .foreach(kv => {
            c += 1
            val clazz: String = kv._1
            val cnt: Int = kv._2

            // 设置参数
            pSt.setString(1, clazz)
            pSt.setInt(2, cnt)

            // 执行插入语句
            //            pSt.execute()
            pSt.addBatch() // 将构建好的insert语句添加到Batch中，等待批量执行

            if (c >= batchSize) {
              // 执行批量插入
              pSt.executeBatch()
              c = 0
            }
          })
        if (c > 0) {
          // 执行批量插入
          pSt.executeBatch()
        }
      })


  }

}
